package MapReduce.Demo6_PhoneTrafficSort;

import MapReduce.writableBean.PhoneTrafficWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import utils.JobSubmit;

import java.io.IOException;

/**
 * @Author lixinlei
 * @Date 2023/3/20 10:17
 */
public class PhoneTrafficSortApp {

    public static class PTSMapper extends Mapper<LongWritable, Text, PhoneTrafficWritable,Text>{


        PhoneTrafficWritable outKey = new PhoneTrafficWritable();
        Text outValue = new Text();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, PhoneTrafficWritable, Text>.Context context) throws IOException, InterruptedException {

            String line = value.toString();

            String[] fields = line.split("\t");

            outKey.setUpFlow(Integer.parseInt(fields[1]));
            outKey.setDownFlow(Integer.parseInt(fields[2]));
            outKey.setSumFlow(Integer.parseInt(fields[3]));
            outValue.set(fields[0]);

            context.write(outKey,outValue);
        }
    }

    public static class PTSReducer extends Reducer<PhoneTrafficWritable,Text,Text, PhoneTrafficWritable>{
        @Override
        protected void reduce(PhoneTrafficWritable key, Iterable<Text> values, Reducer<PhoneTrafficWritable, Text, Text, PhoneTrafficWritable>.Context context) throws IOException, InterruptedException {

            /**
             * 数据格式
             *      上行流量 下行流量 总流量   Iterable(手机号1,手机号2,...)
             */
            for (Text value : values) {
                //相同的流量为key，可能有多个手机号作为value，但是只要遍历每个手机号的时候把流量信息带出来即可
                context.write(value,key);
            }
        }
    }

    public static void main(String[] args) {

        JobSubmit.submitBaseJob(
                PhoneTrafficSortApp.class,
                args,
//                "sort",
                PhoneTrafficSortComparator.class);

    }

}
